package com.my.study.rocketmq.delay;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.TimeUnit;

/**
 * 生产者（延迟消息）
 *
 * @author yyb8152
 * @since 20220509
 **/
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("39.101.165.196:9876");
        //3.启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message message = new Message("DelayTopic", "Tag1", ("Hello rocketMQ" + i).getBytes());

            /**
             * 设置消息延迟发送时间(级别2：:5s)
             * // org/apache/rocketmq/store/config/MessageStoreConfig.java
             * private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
             */
            message.setDelayTimeLevel(3);

            //5.发送同步消息
            SendResult result = producer.send(message);
            // 发送状态
            SendStatus sendStatus = result.getSendStatus();
            // 消息id
            String msgId = result.getMsgId();
            // 消息接受队列ID
            int messageQueueId = result.getMessageQueue().getQueueId();

            System.out.println("发送结果：" + result);
            //System.out.println("发送状态=" + sendStatus + "," + "消息id=" + msgId + "," + "消息队列id=" + messageQueueId);

            // 休眠1s
            TimeUnit.SECONDS.sleep(1);

        }

        //6.关闭生产者producer
        producer.shutdown();
    }

}
